package com.fine.spark.connector.base.worker.process

import com.fine.spark.connector.base.beans.BaseConfig
import com.fine.spark.connector.base.beans.process.SQLProcessConfig
import com.fine.spark.connector.base.worker.BaseWorker
import org.apache.spark.sql.SparkSession

/**
 * @author jayce
 * @date 2021/11/21 2:50 PM
 * @version 1.0
 */
class SQLProcessWorker extends BaseWorker {

  /**
   * 定义处理过程的方法
   *
   * @param bean
   * @param ss
   */
  override def process(config: BaseConfig)(implicit ss: SparkSession): Unit = {
    val sqlConfig: SQLProcessConfig = config.asInstanceOf[SQLProcessConfig]
    val code: String = sqlConfig.code
    logger.info(s"sql script:${System.lineSeparator()}$code")
    ss.sql(code).createOrReplaceTempView(sqlConfig.viewName)
    this.afterProcess(sqlConfig)
  }
}


object SQLProcessWorker {
  def apply: SQLProcessWorker = new SQLProcessWorker()
}